package cn.edu.flink.scala.tutorial.window

import cn.edu.flink.scala.tutorial.source.{ClickSource, Event}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{TumblingEventTimeWindows, TumblingProcessingTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

object WindowTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val eventsDS = env.addSource(new ClickSource())

    // 按键分区
    val windowedStream: WindowedStream[Event, String, TimeWindow] = eventsDS
      .keyBy(_.user)
      .window(TumblingProcessingTimeWindows.of(Time.seconds(1L)))

    // 非按键分区  windowAll 开窗后并行度变为1，所以不推荐使用，使用方法和 window 相似
    val allWindowedStream: AllWindowedStream[Event, TimeWindow] = eventsDS
      .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(1L)))

    // 配置分区策略 配置分区策略后才能使用 EventTimeWindows
    val eventsDS2 = eventsDS.assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps()
      .withTimestampAssigner(
        new SerializableTimestampAssigner[Event] {
          override def extractTimestamp(t: Event, l: Long): Long = t.timestamp
        }))


    eventsDS2.map(event => (event.user, 1L))
      .keyBy(_._1)
      .window(TumblingEventTimeWindows.of(Time.seconds(5)))  // 基于事件时间的滚动窗口
//      .window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8))) // 基于处理时间的滚动窗口
//      .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(10))) // 基于事件时间的滑动窗口
//      .window(EventTimeSessionWindows.withGap(Time.seconds(10))) // 基于事件时间的会话窗口
//      .countWindow(10, 2) // 滑动计数窗口
//      .sum(1)  // sum 聚合
      .reduce((state, data) => (state._1, state._2 + data._2))  //增量聚合
      .print()

    env.execute("WindowTest")

  }
}
